package com.heisen.rocketmq.utils;



import com.heisen.base.utils.SpringUtils;
import com.heisen.rocketmq.model.BaseMessage;
import com.heisen.rocketmq.producer.RocketMQProducer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

/**
 * @author sen
 * @description
 * @date 2023/8/8 11:26
 */
@Slf4j
public class RocketMQUtils {


    private static RocketMQProducer rocketMQTemplate = SpringUtils.getBean(RocketMQProducer.class);

    private static String topicJoinTags(String topic, String tags) {
        return String.format("%s:%s", topic, tags);
    }

    /**
     * 发送普通消息
     * 将消息直接发送到MQ中 ,不管是否被消费 ,只负责单纯的发送消息
     *
     * @param message 消息信息
     * @param topic   主题
     */
    public static void sendMessage(String topic, BaseMessage message) {
        Message sendMessage = MessageBuilder.withPayload(message)
                //设置消息KEYS,一般是数据的唯一ID,主要用于在仪表盘中方便搜索
                .setHeader(RocketMQHeaders.KEYS, message.getId())
                .build();
        rocketMQTemplate.convertAndSend(topic, sendMessage);
    }

    /**
     * 发送普通消息
     *
     * @param data  消息信息
     * @param topic 主题
     * @param tags  主题的标签
     */
    public static void sendMessage(String topic, String tags, BaseMessage data) {
        sendMessage(topicJoinTags(topic, tags), data);
    }

    /**
     * 发送单向信息
     * 不需要等待Broker的响应，直接返回。这种方式适用于不需要关注消息发送结果的场景，如日志记录、统计信息等
     *
     * @param topic   主题
     * @param message 消息信息
     */
    public static void sendOneWay(String topic, BaseMessage message) {
        Message sendMessage = MessageBuilder.withPayload(message)
                //设置消息KEYS,一般是数据的唯一ID,主要用于在仪表盘中方便搜索
                .setHeader(RocketMQHeaders.KEYS, message.getId())
                .build();
        rocketMQTemplate.send(topic, sendMessage);
    }

    /**
     * 发送单向信息
     *
     * @param topic   主题
     * @param tags    主题的标签
     * @param message 信息
     */
    public static void sendOneWay(String topic, String tags, BaseMessage message) {
        sendOneWay(topicJoinTags(topic, tags), message);
    }

    /**
     * 同步发送消息
     *
     * @param topic   主题
     * @param message 主题
     */
    public static SendResult sendSync(String topic, BaseMessage message) {
        log.info("【同步发送消息】 {}", topic);
        Message sendMessage = MessageBuilder.withPayload(message)
                //设置消息KEYS,一般是数据的唯一ID,主要用于在仪表盘中方便搜索
                .setHeader(RocketMQHeaders.KEYS, message.getId())
                .build();
        return rocketMQTemplate.syncSend(topic, sendMessage);
    }

    /**
     * 同步发送消息
     *
     * @param topic   主题
     * @param tags    主题的标签
     * @param message 信息
     * @return
     */
    public static SendResult sendSync(String topic, String tags, BaseMessage message) {
        return sendSync(topicJoinTags(topic, tags), message);
    }

    /**
     * 发布异步信息
     *
     * @param topic
     * @param message
     * @return
     */
    public static void sendAsync(String topic, BaseMessage message, SendCallback sendCallback) {
        Message sendMessage = MessageBuilder.withPayload(message)
                //设置消息KEYS,一般是数据的唯一ID,主要用于在仪表盘中方便搜索
                .setHeader(RocketMQHeaders.KEYS, message.getId())
                .build();
        rocketMQTemplate.asyncSend(topic, message, sendCallback);
    }

    /**
     * 发布异步信息
     *
     * @param topic
     * @param tags
     * @param message
     * @param sendCallback
     */
    public static void sendAsync(String topic, String tags, BaseMessage message, SendCallback sendCallback) {
        sendAsync(topicJoinTags(topic, tags), message, sendCallback);
    }

    /**
     * 发布延迟消息
     *
     * @param topic
     * @param message
     * @param delayLevel
     * @return
     */
    public static SendResult sendDelayed(String topic, BaseMessage message, int delayLevel) {
        return rocketMQTemplate.syncSendDelayTimeSeconds(topic, message, delayLevel);
    }

    public static SendResult sendDelayed(String topic, String tags, BaseMessage message, int delayLevel) {
        return sendDelayed(topicJoinTags(topic, tags), message, delayLevel);
    }


}

